package com.lxp.design.shop.order.config;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.transaction.KafkaTransactionManager;

import java.util.Map;

/**
 * @author LaiXiaoPeng
 * @version 1.0
 * @date 2022/9/14 15:50
 */
@Configuration
@EnableKafka
public class KafkaConfig {
    //配置文件读取
    @Bean("kafkaProperties")
    @Primary
    @ConfigurationProperties(prefix = "spring.kafka")
    public KafkaProperties kafkaProperties() {
        return new KafkaProperties();
    }

    //消费者配置
    @Bean("kafkaListenerFactory")
    @Primary
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(1);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

    private ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    private Map<String, Object> consumerConfigs() {
        Map<String, Object> properties = kafkaProperties().buildConsumerProperties();
        //此处配置非公共配置
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return properties;
    }

    //生产者配置
    @Bean("kafkaTemplate")
    @Primary
    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> factory) {
        return new KafkaTemplate<>(factory);
    }

    @Bean("factory")
    public ProducerFactory<String, String> producerFactory() {
        DefaultKafkaProducerFactory<String, String> factory = new DefaultKafkaProducerFactory<>(producerConfig());
        factory.transactionCapable();
        //此处配置非公共配置
        factory.setKeySerializer(new StringSerializer());
        factory.setValueSerializer(new StringSerializer());
        factory.setTransactionIdPrefix("shop-transaction_");
        return factory;
    }

    private Map<String, Object> producerConfig() {
        Map<String, Object> properties = kafkaProperties().buildProducerProperties();
        //开启幂等性,可保证单分区数据有序性和防止重复发送消息
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        return properties;
    }

    @Bean
    public KafkaTransactionManager<String, String> transactionManager(ProducerFactory<String, String> factory) {
        return new KafkaTransactionManager<>(factory);
    }

    //管理客户端配置
    @Bean("kafkaAdminClient")
    @Primary
    public AdminClient kafkaAdminClient() {
        return AdminClient.create(adminConfigs());
    }

    private Map<String, Object> adminConfigs() {
        return kafkaProperties().buildAdminProperties();
    }
}
